package drds.binlog.common.zookeeper;

import org.apache.commons.lang.StringUtils;

import java.text.MessageFormat;

/**
 * 存储结构：
 * <p>
 * <pre>
 * /otter
 *    canal
 *      cluster
 *      destinations
 *        dest1
 *          running (EPHEMERAL)
 *          cluster
 *          client1
 *            running (EPHEMERAL)
 *            cluster
 *            filter
 *            cursor
 *            mark
 *              1
 *              2
 *              3
 * </pre>
 */
public class Zookeepers
{

    public static final String ZOOKEEPER_SEPARATOR = "/";

    public static final String OTTER_ROOT_NODE = ZOOKEEPER_SEPARATOR + "otter";

    public static final String CANAL_ROOT_NODE = OTTER_ROOT_NODE + ZOOKEEPER_SEPARATOR
            + "canal";

    public static final String DESTINATION_ROOT_NODE = CANAL_ROOT_NODE + ZOOKEEPER_SEPARATOR
            + "destinations";

    public static final String FILTER_NODE = "filter";

    public static final String BATCH_MARK_NODE = "mark";

    public static final String PARSE_NODE = "parse";

    public static final String CURSOR_NODE = "cursor";

    public static final String RUNNING_NODE = "running";

    public static final String CLUSTER_NODE = "cluster";

    public static final String DESTINATION_NODE = DESTINATION_ROOT_NODE
            + ZOOKEEPER_SEPARATOR + "{0}";

    public static final String DESTINATION_PARSE_NODE = DESTINATION_NODE + ZOOKEEPER_SEPARATOR
            + PARSE_NODE;

    public static final String DESTINATION_CLIENTID_NODE = DESTINATION_NODE + ZOOKEEPER_SEPARATOR
            + "{1}";

    public static final String DESTINATION_CURSOR_NODE = DESTINATION_CLIENTID_NODE
            + ZOOKEEPER_SEPARATOR + CURSOR_NODE;

    public static final String DESTINATION_CLIENTID_FILTER_NODE = DESTINATION_CLIENTID_NODE
            + ZOOKEEPER_SEPARATOR + FILTER_NODE;

    public static final String DESTINATION_CLIENTID_BATCH_MARK_NODE = DESTINATION_CLIENTID_NODE
            + ZOOKEEPER_SEPARATOR + BATCH_MARK_NODE;

    public static final String DESTINATION_CLIENTID_BATCH_MARK_WITH_ID_PATH = DESTINATION_CLIENTID_BATCH_MARK_NODE
            + ZOOKEEPER_SEPARATOR + "{2}";

    /**
     * 服务端当前正在提供服务的running节点
     */
    public static final String DESTINATION_RUNNING_NODE = DESTINATION_NODE + ZOOKEEPER_SEPARATOR
            + RUNNING_NODE;

    /**
     * 客户端当前正在工作的running节点
     */
    public static final String DESTINATION_CLIENTID_RUNNING_NODE = DESTINATION_CLIENTID_NODE
            + ZOOKEEPER_SEPARATOR + RUNNING_NODE;

    /**
     * 整个canal server的集群列表
     */
    public static final String CANAL_CLUSTER_ROOT_NODE = CANAL_ROOT_NODE + ZOOKEEPER_SEPARATOR
            + CLUSTER_NODE;

    public static final String CANAL_CLUSTER_NODE = CANAL_CLUSTER_ROOT_NODE
            + ZOOKEEPER_SEPARATOR + "{0}";

    /**
     * 针对某个destination的工作的集群列表
     */
    public static final String DESTINATION_CLUSTER_ROOT = DESTINATION_NODE + ZOOKEEPER_SEPARATOR
            + CLUSTER_NODE;
    public static final String DESTINATION_CLUSTER_NODE = DESTINATION_CLUSTER_ROOT
            + ZOOKEEPER_SEPARATOR + "{1}";

    public static String getDestinationPath(String destinationName)
    {
        return MessageFormat.format(DESTINATION_NODE, destinationName);
    }

    public static String getClientIdNodePath(String destinationName, short clientId)
    {
        return MessageFormat.format(DESTINATION_CLIENTID_NODE, destinationName, String.valueOf(clientId));
    }

    public static String getFilterPath(String destinationName, short clientId)
    {
        return MessageFormat.format(DESTINATION_CLIENTID_FILTER_NODE, destinationName, String.valueOf(clientId));
    }

    public static String getBatchMarkPath(String destinationName, short clientId)
    {
        return MessageFormat.format(DESTINATION_CLIENTID_BATCH_MARK_NODE, destinationName, String.valueOf(clientId));
    }

    public static String getBatchMarkWithIdPath(String destinationName, short clientId, Long batchId)
    {
        return MessageFormat.format(DESTINATION_CLIENTID_BATCH_MARK_WITH_ID_PATH,
                destinationName,
                String.valueOf(clientId),
                getBatchMarkNode(batchId));
    }

    public static String getCursorPath(String destination, short clientId)
    {
        return MessageFormat.format(DESTINATION_CURSOR_NODE, destination, String.valueOf(clientId));
    }

    public static String getCanalClusterNode(String node)
    {
        return MessageFormat.format(CANAL_CLUSTER_NODE, node);
    }

    /**
     * 服务端当前正在提供服务的running节点
     */
    public static String getDestinationServerRunning(String destination)
    {
        return MessageFormat.format(DESTINATION_RUNNING_NODE, destination);
    }

    /**
     * 客户端当前正在工作的running节点
     */
    public static String getDestinationClientRunning(String destination, short clientId)
    {
        return MessageFormat.format(DESTINATION_CLIENTID_RUNNING_NODE, destination, String.valueOf(clientId));
    }

    public static String getDestinationClusterNode(String destination, String node)
    {
        return MessageFormat.format(DESTINATION_CLUSTER_NODE, destination, node);
    }

    public static String getDestinationClusterRoot(String destination)
    {
        return MessageFormat.format(DESTINATION_CLUSTER_ROOT, destination);
    }

    public static String getParsePath(String destination)
    {
        return MessageFormat.format(DESTINATION_PARSE_NODE, destination);
    }

    /**
     * 将batchNode转换为Long
     */
    public static short getClientId(String clientNode)
    {
        return Short.valueOf(clientNode);
    }

    /**
     * 将batchNode转换为Long
     */
    public static long getBatchMarkId(String batchMarkNode)
    {
        return Long.valueOf(batchMarkNode);
    }

    /**
     * 将batchId转化为zookeeper中的node名称
     */
    public static String getBatchMarkNode(Long batchId)
    {
        return StringUtils.leftPad(String.valueOf(batchId.intValue()), 10, '0');
    }
}
